package com.rr.learningdemo.rabbitmq.listener;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rr.learningdemo.rabbitmq.dto.MessageWithLevel;
import com.rr.learningdemo.rabbitmq.thread.RedisListenerThread;
import com.rr.learningdemo.rabbitmq.thread.RedisListenerThreadWithLevel;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;

/**
 * @author zhuangcanrun
 * @Date 2021/06/21/16:51
 * @description
 **/

@Component
public class RabbitMQListenerForMessageWithLevel {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedisListenerThreadWithLevel redisListenerThreadWithLevel;

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = "queueForDistinctWithLevel", durable = "true"),
                    exchange = @Exchange(value = "exchangeForDistinct", type = ExchangeTypes.DIRECT, durable = "true"),
                    key = "queueForDistinctWithLevel"
            )
    )
    private void onMessage(Message message, Channel channel) throws IOException {
        String newMsg = new String(message.getBody(), StandardCharsets.UTF_8);
        List<MessageWithLevel> messages = JSON.parseArray(newMsg, MessageWithLevel.class);
        ZSetOperations zSetOperations = redisTemplate.opsForZSet();
        for (MessageWithLevel m : messages) {
            String key = m.getKey();
            if (zSetOperations.score("zset", key) == null) {
                System.out.println("插入:" + key);
                zSetOperations.add("zset", key, m.getLevel());
            } else {
                System.out.println(key + "已存在队列");
            }
        }
        redisListenerThreadWithLevel.resumeThread();
    }
}
